package com.archstudy.springcloud.message.service.impl.messageDispatch;

import com.archstudy.springcloud.message.service.impl.MessagePushServiceImpl;
import com.lmax.disruptor.RingBuffer;
import com.lmax.disruptor.YieldingWaitStrategy;
import com.lmax.disruptor.dsl.Disruptor;
import com.lmax.disruptor.dsl.ProducerType;
import lombok.extern.slf4j.Slf4j;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.apache.kafka.clients.consumer.ConsumerRecords;
import org.apache.kafka.clients.consumer.KafkaConsumer;

import java.util.ArrayList;
import java.util.Iterator;
import java.util.List;
import java.util.concurrent.Executor;
import java.util.concurrent.Executors;

@Slf4j
class RelayMessageDispatcher implements Runnable {

    private KafkaConsumer consumer;
    private MsgEventProducer msgEventProducer;
    private MessagePushServiceImpl messagePushService;
    private int handlerCount;


    public RelayMessageDispatcher(KafkaConsumer consumer, MessagePushServiceImpl messagePushService, int handlerCount) {
        this.consumer = consumer;
        this.messagePushService = messagePushService;
        this.handlerCount = handlerCount;
    }

    @Override
    public void run() {
        try {
            initDisruptor();
            while (!Thread.currentThread().isInterrupted()) {
                ConsumerRecords consumerRecords = consumer.poll(1000);
                Iterator<ConsumerRecord> it = consumerRecords.iterator();
                while (it.hasNext()) {
                    ConsumerRecord record = it.next();
                    msgEventProducer.onData(record);
                }
            }
        } catch (Exception e) {
            log.error("Exception happened!", e);
        } finally {
            consumer.close();
        }


    }

    public void initDisruptor() {
        Executor executor = Executors.newCachedThreadPool();
        // 事件工厂
        MsgEventFactory eventFactory = new MsgEventFactory();
        // 环形队列大小，2的指数
        int bufferSize = 1048576;
        // 构造 分裂者 （事件分发者）
        Disruptor disruptor = new Disruptor<MsgEvent>(eventFactory,
                bufferSize, executor, ProducerType.SINGLE, new YieldingWaitStrategy());
        // 连接 消费者 处理器
        //disruptor.handleEventsWith(new LongEventHandler());

        List<MsgEventHandler> handlers = new ArrayList<>();
        for (int i = 0; i < handlerCount; i++) {
            handlers.add(new MsgEventHandler(messagePushService));
        }
        disruptor.handleEventsWithWorkerPool((MsgEventHandler[]) handlers.toArray());
//            disruptor.handleEventsWith(longEventHandlers);
        // 开启 分裂者（事件分发）
        disruptor.start();
        // 获取环形队列，用于生产 事件
        RingBuffer<MsgEvent> ringBuffer = disruptor.getRingBuffer();
        msgEventProducer = new MsgEventProducer(ringBuffer);
    }
}
